package com.canal.main;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.common.utils.AddressUtils;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.CollectionUtils;

import java.net.InetSocketAddress;
import java.util.List;
import java.util.Map;

/**
 * @Description
 * @Author 顾志杰
 * @Date 2022/8/6 20:51
 */
@Slf4j
public class CanalSample {

  public static void main(String[] args) throws InterruptedException {
    CanalConnector connector = CanalConnectors.newSingleConnector(
      new InetSocketAddress(AddressUtils.getHostIp(),11111),"example","","");
       try {
        connector.connect();
        connector.subscribe(".*\\..*");
        connector.rollback();
        while (true){
          //一回取多少数据
          Message message = connector.getWithoutAck(100);
          long batchId = message.getId();
          if (batchId == -1||message.getEntries().size()==0) {
            log.info("没有数据更新!!!!!!!!!");
            Thread.sleep(1000);
          }else{
            printEntry(message.getEntries());
          }
          connector.ack(batchId);
        }
       }finally {
         connector.disconnect();
       }
  }

  private static void printEntry(List<CanalEntry.Entry> entries){
    for(CanalEntry.Entry entry : entries){
      if(entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND){
        continue;
      }
      CanalEntry.RowChange rowChange = null;
      try {
        rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
      }catch (Exception e){
        throw new RuntimeException("data:"+entry.toString(),e);
      }
      CanalEntry.EventType eventType = rowChange.getEventType();
      log.info(String.format("=======================&gt; binlog[%s:%s] , name[%s:%s], eventType : %s",
        entry.getHeader().getLogfileName(),entry.getHeader().getLogfileOffset(),
        entry.getHeader().getSchemaName(),entry.getHeader().getTableName(),
        eventType));

      for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
        log.info("=========================================");
        log.info("event type " + eventType.name());
        log.info("****************************** before");
        printColumn(rowData.getBeforeColumnsList());
        log.info("****************************** after");
        printColumn(rowData.getAfterColumnsList());
      }
    }
  }

  private static void printColumn(List<CanalEntry.Column> columns){
    if(CollectionUtils.isEmpty(columns)){
      return;
    }
    for (CanalEntry.Column column : columns) {
      log.info(column.getName() + ":" +column.getValue() +"   update=" +column.getUpdated());
    }
  }
}
